/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.enodeframework.ons.message

import com.aliyun.openservices.ons.api.Action
import com.aliyun.openservices.ons.api.ConsumeContext
import com.aliyun.openservices.ons.api.Message
import com.aliyun.openservices.ons.api.MessageListener
import com.aliyun.openservices.ons.api.batch.BatchMessageListener
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext
import com.aliyun.openservices.ons.api.order.MessageOrderListener
import com.aliyun.openservices.ons.api.order.OrderAction
import org.enodeframework.common.extensions.SysProperties
import org.enodeframework.queue.MessageHandlerHolder
import org.enodeframework.queue.QueueMessage
import java.util.concurrent.CountDownLatch
import java.util.function.Consumer

/**
 * @author anruence@gmail.com
 */
class OnsMessageListener(private val messageHandlerHolder: MessageHandlerHolder) : MessageListener {
    override fun consume(message: Message, consumeContext: ConsumeContext): Action {
        val queueMessage = covertToQueueMessage(message)
        val messageHandler = messageHandlerHolder.chooseMessageHandler(queueMessage.type)
        val latch = CountDownLatch(1)
        messageHandler.handle(queueMessage) { latch.countDown() }
        latch.await()
        return Action.CommitMessage
    }
}

class OnsMessageOrderListener(private val messageHandlerHolder: MessageHandlerHolder) : MessageOrderListener {
    override fun consume(message: Message, context: ConsumeOrderContext?): OrderAction {
        val queueMessage = covertToQueueMessage(message)
        val messageHandler = messageHandlerHolder.chooseMessageHandler(queueMessage.type)
        val latch = CountDownLatch(1)
        messageHandler.handle(queueMessage) { latch.countDown() }
        latch.await()
        return OrderAction.Success
    }
}



class OnsBatchMessageListener(private val messageHandlerHolder: MessageHandlerHolder) : BatchMessageListener {
    override fun consume(messages: List<Message>, consumeContext: ConsumeContext): Action {
        val latch = CountDownLatch(messages.size)
        messages.forEach(Consumer { msg: Message ->
            val queueMessage = covertToQueueMessage(msg)
            val messageHandler = messageHandlerHolder.chooseMessageHandler(queueMessage.type)
            messageHandler.handle(queueMessage) { latch.countDown() }
        })
        latch.await()
        return Action.CommitMessage
    }
}

private fun covertToQueueMessage(messageExt: Message): QueueMessage {
    val queueMessage = QueueMessage()
    val mType = messageExt.getUserProperties(SysProperties.MESSAGE_TYPE_KEY)
    queueMessage.body = messageExt.getBody()
    queueMessage.type = mType
    queueMessage.topic = messageExt.topic
    queueMessage.tag = messageExt.tag
    queueMessage.routeKey = messageExt.shardingKey
    queueMessage.key = messageExt.key
    return queueMessage
}
